package spring.cloud.tasks.tasks_manager.tasks_manager.service;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.BadVersionException;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.apache.zookeeper.data.Stat;
import spring.cloud.tasks.tasks_manager.tasks_manager.entity.TaskConfiguration;
import spring.cloud.tasks.tasks_manager.tasks_manager.node.Node;
import spring.cloud.tasks.tasks_manager.tasks_manager.utils.CuratorFrameworkUtils;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * This class should be modified, when the curator bug is fixed. The bug is <a href="https://issues.apache.org/jira/browse/CURATOR-430">CURATOR-430</a>
 */
@Slf4j
@Setter
@Getter
public class ExecutorCleanService {
    private CuratorFramework curatorFramework;

    private UpdateTaskConfigService updateTaskConfigService;

    public ExecutorCleanService(CuratorFramework curatorFramework, UpdateTaskConfigService updateTaskConfigService) {
        this.curatorFramework = curatorFramework;
        this.updateTaskConfigService = updateTaskConfigService;
    }

    /**
     * delete $SaturnExecutors/executors/xxx<br> delete $Jobs/job/servers/xxx<br> delete $Jobs/job/config/preferList
     * content about xxx
     */
    public void clean(String executorId) {

        try {
            String executorCleanNodePath = Node.getExecutorCleanNodePath(executorId);
            byte[] bytes = curatorFramework.getData().forPath(executorCleanNodePath);
            if (bytes == null || bytes.length == 0) {
                return;
            }

            String cleanNodeData = new String(bytes, StandardCharsets.UTF_8);
            if (!Boolean.parseBoolean(cleanNodeData)) {
                return;
            }

            if (curatorFramework.checkExists().forPath(Node.getExecutorIpNodePath(executorId)) == null) {
                log.info("Clean the executor {}", executorId);
                // delete $SaturnExecutors/executors/xxx
                deleteExecutor(executorId);
                List<String> taskList = getTaskList();
                for (String taskId : taskList) {
                    // delete $Jobs/job/servers/xxx
                    deleteTaskExecutor(taskId, executorId);
                    // delete $Jobs/job/config/preferList content about xxx
                    String preferList = updateTaskConfigPreferListContentToRemoveDeletedExecutor(taskId, executorId);
                    if (preferList != null) {
                        TaskConfiguration taskConfigInfo = new TaskConfiguration(curatorFramework.getNamespace(), taskId, preferList);
                        List<TaskConfiguration> taskConfigInfoList = new ArrayList<>();
                        taskConfigInfoList.add(taskConfigInfo);
                        updatePreferList(taskConfigInfoList);
                    }
                }
            } else {
                log.info("The executor {} is online now, no necessary to clean", executorId);
            }
        } catch (NoNodeException e) {
            log.debug("No clean node found for executor:" + executorId, e);
        } catch (Exception e) {
            log.error("Clean the executor " + executorId + " error", e);
        } finally {

        }
    }

    private List<String> getTaskList() throws KeeperException.ConnectionLossException, InterruptedException {
        List<String> taskList = new ArrayList<>();
        try {
            String tasksNodePath = Node.TASKS_NODE_PATH;
            if (curatorFramework.checkExists().forPath(tasksNodePath) != null) {
                List<String> list = curatorFramework.getChildren().forPath(tasksNodePath);
                if (list != null && !list.isEmpty()) {
                    taskList.addAll(list);
                }
            }
        } catch (NoNodeException e) { // NOSONAR
            // ignore
        } catch (KeeperException.ConnectionLossException | InterruptedException e) {
            throw e;
        } catch (Exception e) {
            log.error("Clean the executor, getJobList error", e);
        }
        return taskList;
    }

    /**
     * delete $SaturnExecutors/executors/xxx
     */
    private void deleteExecutor(String executorName) throws KeeperException.ConnectionLossException, InterruptedException {
        try {
            String executorNodePath = Node.getExecutorNodePath(executorName);
            CuratorFrameworkUtils.deleteRecursive(curatorFramework, executorNodePath);
        } catch (KeeperException.ConnectionLossException | InterruptedException e) {
            throw e;
        } catch (Exception e) {
            log.error("Clean the executor, delete" + " Executor(" + executorName + ") error", e);
        }
    }

    /**
     * delete $Jobs/job/servers/xxx
     */
    private void deleteTaskExecutor(String taskId, String executorId)
            throws KeeperException.ConnectionLossException, InterruptedException {
        try {
            String taskExecutorPath = Node.getTaskExecutorNodePath(taskId, executorId);
            CuratorFrameworkUtils.deleteRecursive(curatorFramework, taskExecutorPath);
        } catch (KeeperException.ConnectionLossException | InterruptedException e) {
            throw e;
        } catch (Exception e) {
            log.error("Clean the executor, deleteJobServerExecutor(" + taskId + ", " + executorId + ") error", e);
        }
    }

    /**
     * delete $Jobs/job/config/preferList content about xxx
     */
    private String updateTaskConfigPreferListContentToRemoveDeletedExecutor(String taskId, String executorId)
            throws KeeperException.ConnectionLossException, InterruptedException {
        try {
            String taskConfigPreferListPath = Node.getTaskConfigPreferListNodePath(taskId);
            Stat stat = new Stat();
            byte[] bytes = curatorFramework.getData().storingStatIn(stat).forPath(taskConfigPreferListPath);
            if (bytes == null || bytes.length == 0) {
                return null;
            }
            // build the new prefer list string
            StringBuilder sb = new StringBuilder();
            String[] strings = new String(bytes, StandardCharsets.UTF_8).split(",");
            boolean found = false;
            for (String string : strings) {
                if (!string.equals(executorId)) {
                    if (sb.length() > 0) {
                        sb.append(',');
                    }
                    sb.append(string);
                } else {
                    found = true;
                }
            }
            curatorFramework.setData().withVersion(stat.getVersion()).forPath(taskConfigPreferListPath, sb.toString().getBytes(StandardCharsets.UTF_8));
            return found ? sb.toString() : null;
        } catch (NoNodeException | BadVersionException e) { // NOSONAR
            // ignore
        } catch (KeeperException.ConnectionLossException | InterruptedException e) {
            throw e;
        } catch (Exception e) {
            log.error("Clean the executor, updateJobConfigPreferListContentToRemoveDeletedExecutor(" + taskId + ", "
                    + executorId + ") error", e);
        }
        return null;
    }

    private void updatePreferList(List<TaskConfiguration> taskConfigInfoList) {
        try {
            if (updateTaskConfigService != null) {
                updateTaskConfigService.batchUpdatePreferList(taskConfigInfoList);
            }
        } catch (Exception e) {
            log.warn("batchUpdatePreferList  error", e); // just log a warn.
        }
    }
}
